package org.flink.marketing.sink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.flink.marketing.constant.EventTrackingConstant;
import org.flink.marketing.constant.SqlConstant;
import org.flink.marketing.domain.BaseBlackListDO;
import org.flink.marketing.util.CheckUtil;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.util.*;

@Slf4j
public class ClickHouseTestSink extends RichSinkFunction<JSONObject> {
    private Connection connection;
    private PreparedStatement preparedStatement;
    private transient Jedis jedis;
    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try(InputStream inputStream = ClickHouseTestSink.class.getResourceAsStream("/ck.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        init();
        super.open(parameters);
        // 加载JDBC驱动
        Class.forName(properties.getProperty("DriverClass"));
        // 获取数据库连接
        connection = DriverManager.getConnection(properties.getProperty("DriverUrl"),properties.getProperty("User"),properties.getProperty("Password"));

    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
    }

    @Override
    public void invoke(JSONObject jsonValue, Context context) throws Exception {

        if (jsonValue == null || jsonValue.isEmpty()){
            log.info("jsonValue为空："+jsonValue);
            return;
        }
        log.info("ck开始："+JSONObject.toJSONString(jsonValue));

        StringBuffer sbSql = new StringBuffer();
        sbSql.append(SqlConstant.INSERT_INTO).append(SqlConstant.INSERT_TABLE);
        sbSql.append("(");
        List<String> tableColums = new ArrayList<>();
        List<String> fileds = new ArrayList<>();

        Iterator<String> it = jsonValue.keySet().iterator();
        //增加黑名单校验
//        if(checkIpEvent(it,jsonValue)){
//            return;
//        }
        int n=0;
        while(it.hasNext()){
            // 获得key
            String key = it.next();
            String value = jsonValue.getString(key);

//            System.out.println("key: "+key+",value:"+value);
            if (n==0){
                sbSql.append(key);
            }else{
                sbSql.append(",").append(key);
            }
            n++;

            value = value == null?"":value;


            fileds.add(key);
            tableColums.add(value);
        }


        sbSql.append(")");
        //values 分隔
        sbSql.append(SqlConstant.INSERT_VALUE).append("(");

        int length = tableColums.size();
        int t=0;
        for (String s :tableColums){
            if (t+1 == length){
                sbSql.append("?");
            }else {
                sbSql.append("?").append(",");
            }
            t++;
        }
//        sbSql.append("?,").append("?");
        sbSql.append(")");

        String sql = sbSql.toString();
        log.info(sql);

        PreparedStatement preparedStatement = connection.prepareStatement(sql);

        for (int i = 0; i < length; i++) {
            preparedStatement.setString(i + 1, tableColums.get(i));
        }
        int ints = preparedStatement.executeUpdate();
//        preparedStatement.addBatch();

        long startTime = System.currentTimeMillis();
//        int[] ints = preparedStatement.executeBatch();
        connection.commit();
        long endTime = System.currentTimeMillis();
        log.info("批量插入完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints);
    }



}
